package com.hmeter.hadoop.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Map.Entry;
import java.util.NavigableMap;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.util.Bytes;

/**
 * Component name: common
 * Utility class used for:
 * 1. Parsing elements in the list by some delimiter
 * 2. Getting most recent timestamp on HBase record (with possibility to filter some column families from calculation)
 * 3. Building xml content response for all HBase Get operations
 * 4. Closing/disposing HTable object after HBase operations
 * 
 * @author Bakir Jusufbegovic
 * @since 0.1
 * @version 0.1
 */
public class HBaseUtility {
	
	private ArrayList<String> parsedList;
	private StringBuilder xmlContentBuilder = new StringBuilder();
	private Configuration conf = null;
	private HTable hTable = null;
	
	public HTable getHTable()
	{
		return hTable;
	}
	
	public void setXmlContent(String content)
	{
		this.xmlContentBuilder.append(content);
	}
	
	public String getXmlContent()
	{
		return xmlContentBuilder.toString();
	}
	
	public HBaseUtility(String hbaseZooKeeperQuorum, String hbaseTable) throws IOException
	{
		conf = new Configuration();
		conf.set("hbase.zookeeper.quorum", hbaseZooKeeperQuorum);
		hTable = new HTable(conf, hbaseTable);
	}
	
	public void closeHTableObject() throws IOException
	{
		if (hTable != null)
		{
			hTable.close();
		}
	}

	public ArrayList<String> parseElementsToList(String list, String delimiter)
	{
		parsedList = new ArrayList<String>();
		String[] listSplit = list.split(delimiter);
		
		for (int i = 0; i < listSplit.length; i++)
		{
			parsedList.add(listSplit[i]);
		}
		
		return parsedList;
	}
	
	public long getMostRecentTimestamp(String rowKey, ArrayList<String> filterColumnFamiliesList) throws IOException
	{
		long maxTimestamp = Long.MIN_VALUE;
		Get get = new Get(Bytes.toBytes(rowKey));

		Result result = hTable.get(get);

		if (!filterColumnFamiliesList.isEmpty()) {
			for (KeyValue kv : result.list()) {
				byte[] bytes = kv.getFamily();
				String str = new String(bytes);
				if (!filterColumnFamiliesList.contains(str)) {
					if (maxTimestamp < kv.getTimestamp()) {
						maxTimestamp = kv.getTimestamp();
					}
				}
			}
		}			
		return maxTimestamp;
	}
	
	public void buildXmlContent(NavigableMap<String,NavigableMap<String,NavigableMap<String,String>>> allContent)
	{		
		for (String rowKey : allContent.keySet())
		{
			xmlContentBuilder.append(" <row key='"+rowKey+"'>\n");
			
			NavigableMap<String,NavigableMap<String,String>> columnFamilyColumnQualifierContent = allContent.get(rowKey);
			
			for (String columnFamily : columnFamilyColumnQualifierContent.keySet())
			{
				xmlContentBuilder.append("  <columnfamily name='"+columnFamily+"'>\n");
				
				NavigableMap<String,String> columnQualifierContent = columnFamilyColumnQualifierContent.get(columnFamily);
				
				for (Entry<String,String> e : columnQualifierContent.entrySet())
				{
					xmlContentBuilder.append("   <columnqualifier name='"+e.getKey()+"'>"+e.getValue()+"</columnqualifier>\n");
				}
				
				xmlContentBuilder.append("  </columnfamily>\n");
			}			
			xmlContentBuilder.append(" </row>\n");
		}		
	}
}
